package com.yh.csx.etl.util;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.List;
import java.util.function.Consumer;
import java.util.function.Function;

import javax.sql.DataSource;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * SQL工具类
 * 
 * @author Robin.Wang
 * @date 2020-03-20
 *
 */
public class SqlUtils {
    private static final Logger log = LoggerFactory.getLogger(SqlUtils.class);

    /**
     * 通过DS执行sql
     */
    public static Object sqlRS(DataSource ds, String sql, Function<ResultSet, Object> fun) {
        try (Connection conn = ds.getConnection();
                Statement stmt = conn.createStatement(ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY)) {
            stmt.setFetchSize(Integer.MIN_VALUE);
            try (ResultSet rs = stmt.executeQuery(sql)) {
                return fun.apply(rs);
            }
        } catch (Exception e) {
            log.error("sqlRs has error, sql: {} ", sql);
            throw new RuntimeException(e);
        }
    }

    public static Object sqlRS(DataSource ds, String sql, List<Object> values, Function<ResultSet, Object> fun) {
        try (Connection conn = ds.getConnection()) {
            try (PreparedStatement pstmt = conn.prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY,
                    ResultSet.CONCUR_READ_ONLY)) {
                pstmt.setFetchSize(Integer.MIN_VALUE);
                if (values != null) {
                    for (int i = 0; i < values.size(); i++) {
                        pstmt.setObject(i + 1, values.get(i));
                    }
                }
                try (ResultSet rs = pstmt.executeQuery()) {
                    return fun.apply(rs);
                }
            }
        } catch (Exception e) {
            log.error("sqlRs has error, sql: {} ", sql);
            throw new RuntimeException(e);
        }
    }

    /**
     * sql执行获取resultSet
     *
     * @param conn     sql connection
     * @param sql      sql
     * @param consumer 回调方法
     */
    public static void sqlRS(Connection conn, String sql, Consumer<ResultSet> consumer) {
        try (Statement stmt = conn.createStatement(); ResultSet rs = stmt.executeQuery(sql)) {
            consumer.accept(rs);
        } catch (SQLException e) {
            log.error(e.getMessage(), e);
        }
    }

    /**
     * 通过DS执行sql
     */
    public static Boolean sqlExcute(DataSource ds, String sql) {
        try (Connection conn = ds.getConnection();
                Statement stmt = conn.createStatement(ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY)) {
            stmt.setFetchSize(Integer.MIN_VALUE);
            if (stmt.execute(sql)) {
                return true;
            }
            return false;
        } catch (Exception e) {
            log.error("sqlRs has error, sql: {} ", sql);
            throw new RuntimeException(e);
        }
    }

    /**
     * 通过DS执行sql
     */
    public static Boolean sqlExcute(DataSource ds, String sql, boolean isBinlogRecorded) {
        try (Connection conn = ds.getConnection();
                Statement stmt = conn.createStatement(ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY)) {
            if (isBinlogRecorded) {
                stmt.execute("SET sql_log_bin=0;");
            }
            stmt.setFetchSize(Integer.MIN_VALUE);
            if (stmt.execute(sql)) {
                return true;
            }
            return false;
        } catch (Exception e) {
            log.error("sqlRs has error, sql: {} ", sql);
            throw new RuntimeException(e);
        }
    }
}
